package com.gandong8.trace.tool.config.rocketmq;

import com.gandong8.trace.tool.config.GlobalProperties;
import com.gandong8.trace.tool.config.rocketmq.properties.RocketMqConsumerCommonData;
import com.gandong8.trace.tool.config.rocketmq.properties.RocketMqConsumerProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Configuration;

@Configuration
@EnableConfigurationProperties({RocketMqConsumerProperties.class, GlobalProperties.class})
@Slf4j
public class RocketMqConsumerConf {
    @Autowired
    RocketMqConsumerProperties publicProp;
    @Autowired
    GlobalProperties globalProperties;

    private void dealConf(RocketMqConsumerCommonData myProp) {
        if(myProp.getBatchMaxSize() == null) {
            myProp.setBatchMaxSize(publicProp.getBatchMaxSize());
        }
        if(myProp.getGroupName() == null) {
            myProp.setGroupName(publicProp.getGroupName());
        }
        if(myProp.getNamesrvAddr() == null) {
            myProp.setNamesrvAddr(publicProp.getNamesrvAddr());
        }
        if(myProp.getThreadMin() == null) {
            myProp.setThreadMin(publicProp.getThreadMin());
        }
        if(myProp.getThreadMax() == null) {
            myProp.setThreadMax(publicProp.getThreadMax());
        }
        if(myProp.getFromWhere() == null) {
            myProp.setFromWhere(publicProp.getFromWhere());
        }
    }

    /**
     * ConsumeFromLastOffset：从上一次消费的位置开始，默认策略
     * ConsumeFromFirstOffset：从头开始消费，如果消费过会出现重复消费的问题
     * ConsumeFromTimestamp：第一次启动从给定时间戳消费，后续启动接着上一次消费的位置继续 和setConsumeTimestamp()配合使用，默认是半个小时以前
     * @param fromWhere 指定消费的offset形式
     * @return ConsumeFromWhere
     */
    private ConsumeFromWhere parseFromWhere(String fromWhere) {
        if("latest".equals(fromWhere)) {
            return ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;
        }
        if("earliest".equals(fromWhere)) {
            return ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET;
        }
        if("timestamp".equals(fromWhere)) {
            return ConsumeFromWhere.CONSUME_FROM_TIMESTAMP;
        }
        return ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;
    }

    protected DefaultMQPushConsumer createConsumer(RocketMqConsumerCommonData myConf, MessageListenerConcurrently listener) {
        if(!globalProperties.getRocketMqConsumerOpen()) {
            System.err.println("rocketMq consumer not allowed");
            return null;
        }
        dealConf(myConf);
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(myConf.getGroupName());
        //多个NameServer地址用分号隔开
        consumer.setNamesrvAddr(myConf.getNamesrvAddr());
        consumer.setConsumeThreadMin(myConf.getThreadMin());
        consumer.setConsumeThreadMax(myConf.getThreadMax());
        consumer.setConsumeMessageBatchMaxSize(myConf.getBatchMaxSize());
        consumer.registerMessageListener(listener);
        consumer.setConsumeFromWhere(parseFromWhere(myConf.getFromWhere()));
        //consumer.setConsumeTimestamp();
        /*
         * 设置消费模型，集群还是广播，默认为集群,使用相同 Consumer ID 的订阅者属于同一个集群
         * 集群消费：当使用集群消费模式时，MQ 认为任意一条消息只需要被集群内的任意一个消费者处理即可。
         * 广播消费：当使用广播消费模式时，MQ 会将每条消息推送给集群内所有注册过的客户端，保证消息至少被每台机器消费一次。
         */
        consumer.setMessageModel(MessageModel.CLUSTERING);
        try {
            // 设置该消费者订阅的主题和tag，如果订阅该主题下的所有tag，则使用*,
            String[] topicArr = myConf.getTopics().split(";");
            for (String tag : topicArr) {
                String[] tagArr = tag.split("~");
                if(tagArr.length == 1) {
                    consumer.subscribe(tagArr[0], "*");
                } else {
                    consumer.subscribe(tagArr[0], tagArr[1]);
                }
            }
            consumer.start();
            log.info("----rocketMq consumer start with topic:{}----", myConf.getTopics());
        } catch (MQClientException e) {
            log.error("consumer 创建失败!", e);
        }
        return consumer;
    }
}
